package com.hvgroup.modules.sanic.service.impl;

import com.hvgroup.modules.sanic.domain.po.LlmApp;
import com.hvgroup.modules.sanic.domain.po.LlmModelConfigApi;
import com.hvgroup.modules.sanic.domain.qo.CPReqBody;
import com.hvgroup.modules.sanic.repository.*;
import com.hvgroup.modules.sanic.service.ChatPlatformService;
import com.hvgroup.modules.sanic.service.impl.chat.EventData;
import com.hvgroup.modules.sanic.service.impl.chat.EventStrategy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.Optional;

//@SuppressWarnings("all")
@Slf4j
@Service
@RequiredArgsConstructor
public class ChatPlatformServiceImpl implements ChatPlatformService {

    private final EventStrategy eventStrategy;
    private final QaSessionRepository qaSessionRepository;
    private final QaMessageRepository qaMessageRepository;

    private final LlmAppRepository llmAppRepository;
    private final LlmModelConfigRepository llmModelConfigRepository;
    private final LlmModelConfigApiRepository llmModelConfigApiRepository;

    @Override
    public Flux<ServerSentEvent<String>> chatCompletion(CPReqBody qaQuery) {

        Mono<LlmModelConfigApi> configMonoApi = Mono.fromCallable(() ->
                {
                    LlmApp app = llmAppRepository.findFirstByAppId(qaQuery.getLevelId_2())
                            .orElseThrow(() -> new RuntimeException("请选择应用!"));
                    return llmModelConfigApiRepository.findById(app.getPlatformApiId())
                            .orElseThrow(() -> new RuntimeException("请选择应用!"));
                }
        ).subscribeOn(Schedulers.boundedElastic());

        return configMonoApi.flatMapMany(platform -> {

            // 核心代码:
            Flux<EventData> eventDataFlux = eventStrategy.getExecute(platform.getPlatformType())
                    .buildChatApi(platform, qaQuery);

            return eventDataFlux
                    .skip(1) // 跳过初始状态
                    .distinctUntilChanged(EventData::getStep) // 只推送 step 变更
                    .filter(o -> o.getStep() > 0) // 有效业务
                    .delayElements(Duration.ofMillis(500))
                    // .filter(e -> StringUtils.isNotBlank(e.getText())) // 如需再判空可加
                    .map(eventData -> {
                        String data = Optional.ofNullable(eventData)
                                .orElseGet(() -> EventData.builder().text("").build())
                                .getText();
                        return ServerSentEvent.<String>builder()
                                .event("message")
                                .data(data)
                                .build();
                    })
                    .onErrorResume(e -> {
                        log.error("处理 SSE 流时发生错误：{}", e.getMessage(), e);
                        return Flux.just(ServerSentEvent.<String>builder()
                                .event("error")
                                .data("流处理异常：" + e.getMessage())
                                .build());
                    })
                    .concatWith(Mono.just(ServerSentEvent.<String>builder()
                            .event("end")
                            .data("END")
                            .build()));
        });
    }

}
